package com.lingjtx.common.thread.mq;

import com.lingjtx.common.core.util.StringUtil;
import com.lingjtx.common.thread.pooling.ObjectPoolTemplate;
import com.rabbitmq.client.Channel;
import jakarta.annotation.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.Connection;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.core.ParameterizedTypeReference;

import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;

public class RabbitSenderTemplate {
    private static final Logger log = LoggerFactory.getLogger(RabbitSenderTemplate.class);
    private final ObjectPoolTemplate<RabbitTemplate> rabbitTemplatePool;

    public RabbitSenderTemplate(@Qualifier("rabbitObjectPoolTemplate") ObjectPoolTemplate<RabbitTemplate> rabbitTemplatePool) {
        this.rabbitTemplatePool = rabbitTemplatePool;
    }

    // 核心发送方法
    public void convertAndSend(Object object) {
        rabbitTemplatePool.consume(rabbitTemplate -> rabbitTemplate.convertAndSend(object));
    }

    public void convertAndSend(String routingKey, Object object) {
        rabbitTemplatePool.consume(rabbitTemplate -> rabbitTemplate.convertAndSend(routingKey, object));
    }

    public void convertAndSend(String exchange, String routingKey, Object object) {
        rabbitTemplatePool.consume(rabbitTemplate -> rabbitTemplate.convertAndSend(exchange, routingKey, object));
    }

    public void send(String exchange, String routingKey, Object message) {
        convertAndSendInternal(exchange, routingKey, message, null, null);
    }

    public void sendWithCorrelation(String exchange, String routingKey, Object message, String correlationId) {
        convertAndSendInternal(exchange, routingKey, message, null, new CorrelationData(correlationId));
    }

    public void sendWithPostProcessor(String exchange, String routingKey, Object message,
                                      MessagePostProcessor processor) {
        convertAndSendInternal(exchange, routingKey, message, processor, null);
    }

    public void sendWithCallbacks(String exchange, String routingKey, Object message,
                                  CorrelationData correlationData) {
        convertAndSendInternal(exchange, routingKey, message, null, correlationData);
    }

    public void sendPersistent(String exchange, String routingKey, Object message) {
        MessagePostProcessor processor = msg -> {
            msg.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
            return msg;
        };
        convertAndSendInternal(exchange, routingKey, message, processor, null);
    }

    public void sendDelayed(String exchange, String routingKey, Object message, long delayMillis) {
        MessagePostProcessor processor = msg -> {
            msg.getMessageProperties().setDelayLong(delayMillis);
            return msg;
        };
        convertAndSendInternal(exchange, routingKey, message, processor, null);
    }

    //  高级功能
    public <T> T rpcRequest(String exchange, String routingKey, Object request,
                            Class<T> responseType, long timeoutMillis) {
        return rabbitTemplatePool.execute(template -> {
            template.setReplyTimeout(timeoutMillis);
            return template.convertSendAndReceiveAsType(
                    exchange, routingKey, request,
                    ParameterizedTypeReference.forType(responseType)
            );
        });
    }

    public void sendBatchInTransaction(String exchange, String routingKey, List<?> messages) {
        rabbitTemplatePool.consume(template -> {
            template.execute(channel -> {
                channel.txSelect();
                try {
                    messages.forEach(msg ->
                            template.convertAndSend(exchange, routingKey, msg)
                    );
                    channel.txCommit();
                    return messages.size();
                } catch (Exception e) {
                    channel.txRollback();
                    throw new RabbitOperationException("事务回滚", e)
                            .addContext("messageCount", messages.size());
                }
            });
        });
    }

    public CompletableFuture<Void> sendAsync(String exchange, String routingKey, Object message) {
        return rabbitTemplatePool.asyncExecute(template -> {
            template.convertAndSend(exchange, routingKey, message);
            return null;
        });
    }

    // 全链路追踪
    public void sendWithTraceId(String exchange, String routingKey, Object message) {
        String traceId = StringUtil.getUuid();
        MessagePostProcessor processor = msg -> {
            msg.getMessageProperties().setHeader("X-Trace-ID", traceId);
            return msg;
        };
        convertAndSendInternal(exchange, routingKey, message, processor, null);
    }

    //  工具方法
    private void convertAndSendInternal(String exchange, String routingKey, Object message,
                                        @Nullable MessagePostProcessor processor,
                                        @Nullable CorrelationData correlationData) {
        logOperation("SEND", exchange, routingKey, message);

        rabbitTemplatePool.consume(template -> {
            try {
                if (processor != null && correlationData != null) {
                    template.convertAndSend(exchange, routingKey, message, processor, correlationData);
                } else if (processor != null) {
                    template.convertAndSend(exchange, routingKey, message, processor);
                } else if (correlationData != null) {
                    template.convertAndSend(exchange, routingKey, message, correlationData);
                } else {
                    template.convertAndSend(exchange, routingKey, message);
                }
            } catch (AmqpException e) {
                throw new RabbitOperationException("消息发送失败", e)
                        .addContext("exchange", exchange)
                        .addContext("routingKey", routingKey)
                        .addContext("messageType", message.getClass().getName());
            }
        });
    }

    private void logOperation(String action, String exchange, String routingKey, Object message) {
        if (log.isDebugEnabled()) {
            log.debug("{} [{}] -> {} | MessageType: {}",
                    action, exchange, routingKey, message.getClass().getSimpleName());
        }
    }

    //  健康监控
    private volatile boolean cachedHealthStatus = false;
    private volatile long lastHealthCheckTime = 0;
    private static final long HEALTH_CHECK_CACHE_TIME = 30_000; // 30秒缓存

    public boolean isHealthy() {
        // 使用缓存减少检查频率
        if (System.currentTimeMillis() - lastHealthCheckTime < HEALTH_CHECK_CACHE_TIME) {
            return cachedHealthStatus;
        }

        boolean currentStatus = checkConnectionHealth();
        cachedHealthStatus = currentStatus;
        lastHealthCheckTime = System.currentTimeMillis();
        return currentStatus;
    }

    private boolean checkConnectionHealth() {
        return rabbitTemplatePool.execute(template -> {
            CachingConnectionFactory factory = (CachingConnectionFactory) template.getConnectionFactory();

            // 创建测试连接（最终验证）
            try (Connection connection = factory.createConnection()) {
                try (Channel channel = connection.createChannel(false)) {
                    // 执行简单测试命令（如声明临时队列）
                    String testQueue = "health_check_" + StringUtil.getUuid();
                    channel.queueDeclare(testQueue, false, true, true, null);
                    channel.queueDelete(testQueue);
                    return true;
                }
            } catch (Exception e) {
                log.warn("RabbitMQ健康检查失败", e);
                return false;
            }
        });
    }

    public Map<String, Object> poolMetrics() {
        return rabbitTemplatePool.getResourceMonitorMap();
    }

    //  异常类
    public static class RabbitOperationException extends RuntimeException {
        private final Map<String, Object> context = new HashMap<>();

        public RabbitOperationException(String message, Throwable cause) {
            super(message, cause);
        }

        public RabbitOperationException addContext(String key, Object value) {
            context.put(key, value);
            return this;
        }

        public Map<String, Object> getContext() {
            return Collections.unmodifiableMap(context);
        }
    }
}
